package com.zhang.first.day04;

import com.zhang.first.utils.ClickEvent;
import com.zhang.first.utils.ClickEventSource;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 每来一条数据，输出一次这个用户对每个url的访问次数的HashMap
 *
 * @title: MapState
 * @author: zhang
 * @date: 2022/1/17 18:48
 */
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickEventSource())
                .keyBy(r -> r.name)
                .process(new KeyedProcessFunction<String, ClickEvent, String>() {
                    private MapState<String, Long> mapState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Long>(
                                "mapState",
                                Types.STRING,
                                Types.LONG
                        ));
                    }

                    @Override
                    public void processElement(ClickEvent value, KeyedProcessFunction<String, ClickEvent, String>.Context ctx, Collector<String> out) throws Exception {
                        //如果当前用户中的url之前没有访问过，那么计为（url，1L）
                        if (!mapState.contains(value.url)) {
                            mapState.put(value.url, 1L);
                        } else {
                            //如果当前数据中的url之前访问过，统计值+1
                            mapState.put(value.url, mapState.get(value.url) + 1L);
                        }
                        StringBuilder result = new StringBuilder();
                        result.append(ctx.getCurrentKey() + "=>{\n");
                        for (String url : mapState.keys()) {
                            result.append("  " + url + ":" + mapState.get(url) + ";\n");
                        }
                        result.append("}\n");

                        out.collect(result.toString());
                    }
                })
                .print();

        env.execute();
    }
}
